import asyncio
import dataclasses
import enum
import time

from tornado.ioloop import IOLoop, PeriodicCallback
from webargs import ValidationError

from webutils.compat import complex_json
from webutils.layers.connection import connection_proxy
from webutils.random import random_string
from webutils.ratelimit.exceptions import RateLimited
from webutils.string import strip
from webutils.tornado.websocket import WebSocketHandler, callback_wrapper
from webutils.tornado.responses import Response
from pubsub.brokers.rabbitmq import RabbitMQ
from message_broker.const import ActType, UserGroup

from .. import broadcaster, const, services, tasks


@dataclasses.dataclass
class User:
    id: str
    name: str
    role: enum.Enum = dataclasses.field(default=None)


@dataclasses.dataclass
class Device:
    id: str
    name: str
    classroom_id: str = None
    type: enum.Enum = dataclasses.field(default=None)


@dataclasses.dataclass
class UserLogoutMessage:
    uid: str
    nm: str
    act: str = str(ActType.USER_OFFLINE.value)


class BrokerHandler(WebSocketHandler):
    """
    有两种接入中控的方式:
    1. 直连：用户直接通过APP接入中控
    2. 代理接入：用户通过本地中控的消息转发接入中控

    直连模式下，当前WebScoket会存储用户的信息
    代理模式下，当前WebSocket会维护通过本地中控的用户信息

    连接建立时，需要初始化以下信息
    直连用户身份/链接模式 基础信息/也用于订阅频道
    session_id 用于启动消息订阅服务
    """
    HEARTBEAT_INTERVAL = 10
    IDLE_TIMEOUT = 3
    EXCHANGE_NAME = 'base_broker'

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.session.is_registered = False
        self.session.is_ready = False
        # 直连模式下，self.session.user 会被初始化
        # user 的数据类型是 User
        self.session.user = None
        # 代理模式下，self.session.agent 会被初始化
        # agent 的数据类型是 Device
        self.session.agent = None
        # 代理模式下，该值会被更新
        self.session.users = []
        self.broadcaster_app = None

    @property
    def is_proxy_mode(self):
        return self.session.user is None

    @callback_wrapper
    @connection_proxy
    async def open(self, conn, *args, **kwargs):
        # 接口认证
        await self.authorize(conn, *args, **kwargs)
        if self.session.user:
            self.session.user_group = \
                const.UserGroup.from_user_role(self.session.user.role)
        else:
            self.session.user_group = const.UserGroup.DEVICE

        self.initialize_broadcaster()
        await self.handle_duplication_connection(conn)
        IOLoop.current().call_later(
            self.IDLE_TIMEOUT,
            self.check_is_registered,
        )
        await self.broadcaster_app.start_consume(self)
        self.session.is_ready = True

    async def authorize(self, conn, **kwargs):
        # session.id
        # session.user
        raise NotImplementedError()

    async def handle_duplication_connection(self, conn):
        connection = services.ConnectionService(
            conn,
            room_id=self.session.room_id,
            session_id=self.session.id,
        )
        lock = await connection.acquire_lock()
        if not lock:
            self.abort('[open] failed to acquire lock')

    @callback_wrapper
    @connection_proxy
    async def on_message(self, conn, message):
        begin_at = time.time()
        while not self.session.is_ready:
            time_cost = time.time() - begin_at
            if time_cost > self.IDLE_TIMEOUT:
                self.abort(f'not ready after {self.IDLE_TIMEOUT} seconds')
            await asyncio.sleep(1)
        if isinstance(message, str):
            await self.on_text_message(conn, message)
        else:
            await self.on_binary_message(conn, message)

    async def on_text_message(self, conn, message):
        self.logger.info('[on_text_message] %s', strip(message))
        try:
            json_message = complex_json.loads(message)
        except complex_json.JSONDecodeError:
            self.logger.warning('invalid text message: %s', strip(message))
            return
        act_type_value = json_message.get('act', None)
        try:
            act_type = const.ActType(int(act_type_value))
        except (TypeError, ValueError):
            self.logger.warning(f'unknown act: {repr(act_type_value)}')
            return
        if not self.session.is_registered and \
                act_type not in [const.ActType.REGISTER,
                                 const.ActType.LOCAL_REGISTER,
                                 const.ActType.BATCH_REGISTER]:
            self.logger.warning('cannot handle act: %s before registered',
                                act_type)
            return
        handler_name = f'on_act_{act_type.name}'.lower()
        handler = getattr(self, handler_name, None)
        if not handler:
            self.logger.warning('cannot get handler %s', handler_name)
            return
        await handler(json_message, conn)

    async def on_binary_message(self, conn, message):
        pass

    async def check_is_registered(self):
        if not self.session.is_registered:
            self.logger.info('closed because of not registered')
            self.close()

    def on_close(self):
        code = self.close_code
        reason = self.close_reason
        self.logger.info(f'websockt is closed. code: {code} reason: {reason}')
        super().on_close()

    async def heartbeat(self):
        pass

    def start_heartbeat(self):
        assert self.heartbeat_callback is None
        self.heartbeat_callback = PeriodicCallback(
            self.heartbeat, self.HAERTBEAT_INTERVAL * 1000)
        self.heartbeat_callback.start()

    def initialize_broadcaster(self):
        if self.broadcaster_app:
            return
        broadcaster_id = '_'.join([
            self.session.room_id,
            self.session.id,
            random_string(length=4),
        ])
        broker = RabbitMQ(
            self.application.settings['RABBITMQ'],
            broadcaster_id,
            self.log_context,
            exchange_name=self.EXCHANGE_NAME,
        )
        channels_subscribed = {
            'session_id': self.session.id,
            'x-match': 'any',
            'group': self.session.user_group,
        }
        task_meta = {
            'session_id': self.session.id,
        }
        if not self.is_proxy_mode:
            channels_subscribed['user_id'] = self.session.user.id
            task_meta['session_user_id'] = self.session.user.id
        self.broadcaster_app = broadcaster.BrokerApp(
            broker,
            room_id=self.session.room_id,
            tasks=tasks.tasks,
            task_meta=task_meta,
            channels_subscribed=channels_subscribed,
        )

    @connection_proxy
    async def release_resource(self, conn):
        # 下线通知
        if self.session.user:
            logout_message = UserLogoutMessage(
                uid=self.session.user.id,
                nm=self.session.user.name,
            )
            await self.broadcaster_app.send_task(
                'broadcast_logout',
                dataclasses.asdict(logout_message),
                force=True,
                group=UserGroup.ALL,
            )
        if self.broadcaster_app:
            await self.broadcaster_app.stop()
        self.stop_background_tasks()
        if self.session.get('user'):
            user_service = services.UserService(
                conn,
                room_id=self.session.room_id,
            )
            await user_service.logout_user(self.session.user.id)
        elif self.session.get('agent'):
            user_service = services.UserService(
                conn,
                room_id=self.session.room_id,
            )
            proxy_user_service = \
                services.ProxyUserService(conn,
                                          room_id=self.session.room_id,
                                          proxy_id=self.session.id)
            proxy_users = await proxy_user_service.get_users()
            self.logger.debug(f'[proxy ws close]{proxy_users=}')
            for uid in proxy_users:
                await user_service.logout_user(uid)

    async def handle_callback_exception(self, exc):
        if isinstance(exc, RateLimited):
            self.logger.debug('rate limit triggered')
            await self.write_response(
                Response(errcode=1100, errmsg='rate limit'))
            return
        return await super().handle_callback_exception(exc)

    async def get_user_info(self, sender, *, conn=None, extra_keys=None):
        """
            消息需要添加发见人信息
            如果消息来自本地中控，一般都包含该信息
            如果消息来自APP，可以不提供该信息
            """
        user_info = {}
        if sender:
            user_info['id'] = sender['id']
            user_info['name'] = sender['name']
            if 'group' in extra_keys:
                user_service = services.UserService(
                    conn,
                    room_id=self.session.room_id,
                )
                user_info['group'] = await user_service.get_user_group(
                    user_info['id'])
        else:
            if self.is_proxy_mode:
                raise ValidationError(
                    'sender should be provided in proxy mode')
            user_info['id'] = self.session.user.id
            user_info['name'] = self.session.name
            if 'group' in extra_keys:
                user_info['group'] = self.session.user_group
        return user_info
